/**
* Copyright (c) By zengqh.
*
* This program is just for fun or demo, in the hope that it  
* will be useful, you can redistribute it and/or modify freely.
*
* Time: 2013/05/29
* File: enn_work_queue.cpp
* Blog: http://www.cnblogs.com/zengqh/
**/

#include "enn_work_queue.h"

namespace enn
{

class WorkerThread : public Noncopyable, public AllocatedObject, public Thread
{
public:
	/// Construct.
	WorkerThread(WorkQueue* owner, unsigned index) :
	  owner_(owner),
		  index_(index)
	  {
	  }

	  /// Process work items until stopped.
	  virtual void ThreadFunction()
	  {
		  owner_->ProcessItems(index_);
	  }

	  /// Return thread index.
	  unsigned GetIndex() const { return index_; }

private:
	/// Work queue.
	WorkQueue* owner_;
	/// Thread index.
	unsigned index_;
};

WorkQueue::WorkQueue() 
	: shutDown_(false),
	pausing_(false),
	paused_(false)
{

}

WorkQueue::~WorkQueue()
{
	// Stop the worker threads. First make sure they are not waiting for work items
	shutDown_ = true;
	Resume();

	for (unsigned i = 0; i < threads_.Size(); ++i)
		threads_[i]->Stop();
}

void WorkQueue::CreateThreads(unsigned numThreads)
{
	// Other subsystems may initialize themselves according to the number of threads.
	// Therefore allow creating the threads only once, after which the amount is fixed
	if (!threads_.Empty())
		return;

	// Start threads in paused mode
	Pause();

	for (unsigned i = 0; i < numThreads; ++i)
	{
		WorkerThread* thread(new WorkerThread(this, i + 1));
		thread->Start();
		threads_.Push(thread);
	}
}

void WorkQueue::AddWorkItem(const WorkItem& item)
{
	// Push to the main thread list to keep item alive
	// Clear completed flag in case item is reused
	workItems_.push_back(item);
	WorkItem* itemPtr = workItems_.back();
	itemPtr->completed_ = false;

	// Make sure worker threads' list is safe to modify
	if (threads_.Size() && !paused_)
		queueMutex_.Acquire();

	// Find position for new item
	if (queue_.Empty())
	{
		queue_.Push(itemPtr);
	}
	else
	{
		for (enn::list<WorkItem*>::type::iterator i = queue_.begin(); i != queue_.end(); ++i)
		{
			if ((*i)->priority_ <= itemPtr->priority_)
			{
				queue_.insert(i, itemPtr);
				break;
			}
		}
	}

	if (threads_.Size())
	{
		queueMutex_.Release();
		paused_ = false;
	}
}

void WorkQueue::Pause()
{
	if (!paused_)
	{
		pausing_ = true;

		queueMutex_.Acquire();
		paused_ = true;

		pausing_ = false;
	}
}

void WorkQueue::Resume()
{
	if (paused_)
	{
		queueMutex_.Release();
		paused_ = false;
	}
}

void WorkQueue::Complete(unsigned priority)
{
	if (threads_.size())
	{
		Resume();

		// Take work items also in the main thread until queue empty or no high-priority items anymore
		while (!queue_.Empty())
		{
			queueMutex_.Acquire();
			if (!queue_.empty() && (queue_.front())->priority_ >= priority)
			{
				WorkItem* item = queue_.front();
				queue_.pop_front();
				queueMutex_.Release();
				item->workFunction_(item, 0);
				item->completed_ = true;
			}
			else
			{
				queueMutex_.Release();
				break;
			}
		}

		// Wait for threaded work to complete
		while (!IsCompleted(priority))
		{
		}

		// If no work at all remaining, pause worker threads by leaving the mutex locked
		if (queue_.empty())
			Pause();
	}
	else
	{
		// No worker threads: ensure all high-priority items are completed in the main thread
		while (!queue_.empty() && queue_.front()->priority_ >= priority)
		{
			WorkItem* item = queue_.front();
			queue_.pop_front();
			item->workFunction_(item, 0);
			item->completed_ = true;
		}
	}

	PurgeCompleted();
}


bool WorkQueue::IsCompleted(unsigned priority) const
{
	for (enn::list<WorkItem>::type::const_iterator i = workItems_.begin(); i != workItems_.end(); ++i)
	{
		if (i->priority_ >= priority && !i->completed_)
			return false;
	}

	return true;
}

void WorkQueue::ProcessItems(unsigned threadIndex)
{
	bool wasActive = false;

	for (;;)
	{
		if (shutDown_)
			return;

		if (pausing_ && !wasActive)
			Time::Sleep(0);
		else
		{
			queueMutex_.Acquire();
			if (!queue_.Empty())
			{
				wasActive = true;

				WorkItem* item = queue_.Front();
				queue_.pop_front();
				queueMutex_.Release();
				item->workFunction_(item, threadIndex);
				item->completed_ = true;
			}
			else
			{
				wasActive = false;

				queueMutex_.Release();
				Time::Sleep(0);
			}
		}
	}
}

void WorkQueue::PurgeCompleted()
{
	
}

}